package com.diamond.smart.data.service.impl;

import com.diamond.smart.data.config.MqttClientConfig;
import com.diamond.smart.data.handler.MqttHandler;
import com.diamond.smart.data.service.MqttClientService;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;

@ConditionalOnProperty(value = "mqtt.enable", havingValue = "true")
@Service
public class MqttClientServiceImpl implements MqttClientService {

    private static final Logger logger = LoggerFactory.getLogger(MqttClientService.class);

    @Autowired
    MqttClientConfig clientConfig;

    @PostConstruct
    public void init() {
        this.connect(clientConfig.getUrl(), clientConfig.getClientId(), clientConfig.getUsername(), clientConfig.getPassword(), clientConfig.getTimeout(), clientConfig.getKeepalive());
        subscribe("device/voltage/1650790956459");
        subscribe("device/4/6/3/BD1640227430104");
    }

    /**
     * 连接mqtt服务器
     *
     * @param url：服务器地址
     * @param clientID：客户端身份标识    唯一
     * @param username：服务器连接用户名
     * @param password：服务器连接密码
     * @param timeout：服务器连接超时时间
     * @param keepalive：服务器连接保活时间
     */
    @Override
    public void connect(String url, String clientID, String username, String password, int timeout, int keepalive) {
        MqttClient client;
        try {
            client = new MqttClient(url, clientID, new MemoryPersistence());
            //设置Mqtt可选设置
            MqttConnectOptions options = new MqttConnectOptions();
            //设置为false则每次都以新身份链接
            options.setCleanSession(false);
            if (StringUtils.hasLength(username)) {
                options.setUserName(username);
            }
            if (StringUtils.hasLength(password)) {
                options.setPassword(password.toCharArray());
            }
            options.setConnectionTimeout(timeout);
            options.setKeepAliveInterval(keepalive);
            clientConfig.setClient(client);
            try {
                clientConfig.getClient().setCallback(new MqttHandler());
                clientConfig.getClient().connect(options);
                logger.info("MQTT连接成功！");
            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 发布消息
     * QoS0 代表，Sender 发送的一条消息，Receiver 最多能收到一次，也就是说 Sender 尽力向 Receiver 发送消息，
     * 如果发送失败，也就算了；
     * <p>
     * QoS1 代表，Sender 发送的一条消息，Receiver 至少能收到一次，也就是说 Sender 向 Receiver 发送消息，
     * 如果发送失败，会继续重试，直到 Receiver 收到消息为止，但是因为重传的原因，Receiver 有可能会收到重复的消息；
     * <p>
     * QoS2 代表，Sender 发送的一条消息，Receiver 确保能收到而且只收到一次，
     * 也就是说 Sender 尽力向 Receiver 发送消息，如果发送失败，会继续重试，直到 Receiver 收到消息为止，
     * 同时保证 Receiver 不会因为消息重传而收到重复的消息。
     *
     * @param qos
     * @param retained:   消息是否持久化到Message
     * @param topic
     * @param pushMessage
     */
    @Override
    public void publish(int qos, boolean retained, String topic, String pushMessage) {
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        pushMessage = "{" + pushMessage + "}";
        message.setPayload(pushMessage.getBytes());
        MqttTopic mTopic = clientConfig.getClient().getTopic(topic);
        if (null == mTopic) {
            logger.error("topic not exist");
        }
        MqttDeliveryToken token;
        try {
            token = mTopic.publish(message);
            token.waitForCompletion();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 默认qos=0，消息不持久化
     *
     * @param topic
     * @param pushMessage
     */
    @Override
    public void publish(String topic, String pushMessage) {
        publish(0, false, topic, pushMessage);
    }

    /**
     * 订阅对应qos等级的主题消息
     *
     * @param topic
     * @param qos
     */
    @Override
    public void subscribe(String topic, int qos) {
        try {
            clientConfig.getClient().subscribe(topic, qos);
            logger.info("订阅成功！");
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 默认qos=0
     *
     * @param topic
     */
    @Override
    public void subscribe(String topic) {
        subscribe(topic, 0);
    }
}
